package org.swift.mashup.engine.deployment;

import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.deployment.Deployer;
import org.apache.axis2.deployment.DeploymentException;
import org.apache.axis2.deployment.repository.util.DeploymentFileData;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class ConcurrentMashupDeployer implements Deployer {
    private static Logger log = Logger.getLogger(ConcurrentMashupDeployer.class);
    private static final int MAX_WORKERS
            = Integer.parseInt(System.getProperty("org.swift.mashup.engine.deployment.ConcurrentMashupDeployer.maxWorkers", "4"), 10);

   	private String directory = null;
    private String extension = null;
   	private ConfigurationContext configCtx = null;
    private BlockingQueue<DeploymentFileData> deploymentQueue = new LinkedBlockingQueue<DeploymentFileData>();
    private MashupDeployer undeployer = null;
    private Collection<Worker> workers = new ArrayList<Worker>();
    private Manager manager = null;

    @Override
    public void deploy(DeploymentFileData deploymentFileData) throws DeploymentException {
        if ( manager == null ) {
            manager = new Manager();
            manager.start();
        }

        deploymentQueue.add(deploymentFileData);
        log.debug(String.format("Added %s to deployment queue", deploymentFileData.getName()));
    }

    @Override
    public void init(ConfigurationContext configurationContext) {
   		this.configCtx = configurationContext;
  	}

    @Override
    public void setDirectory(String directory) {
        this.directory = directory;
    }

    @Override
    public void setExtension(String ext) {
        this.extension = ext;
    }

    @Override
    public void unDeploy(String fileName) throws DeploymentException {
        if ( undeployer == null ) {
            undeployer = newDeployer();
        }
        undeployer.unDeploy(fileName);
    }

    private MashupDeployer newDeployer() {
        MashupDeployer deployer = new MashupDeployer();
        deployer.init(configCtx);
        deployer.setDirectory(directory);
        deployer.setExtension(extension);
        return deployer;
    }

    private class Manager extends Thread {
        private long start = 0;

        private Manager() {
            Runtime.getRuntime().addShutdownHook(new Thread() {
                public void run() {
                    Manager.this.interrupt();
                    for ( Worker worker : workers ) {
                        worker.shutdown();
                    }
                }
            });
        }

        @Override
        public void run() {
            log.info("Manager entering run");

            try {
                while ( !isInterrupted() ) {
                    printQueueStats();
                    adjustWorkerCount();
                    Thread.sleep(10*1000);
                }
            }
            catch (InterruptedException e) {
            }

            log.info("Manager exiting run");
        }

        private void adjustWorkerCount() {
            int workerCount = workers.size();
            boolean create = false;
            boolean delete = false;

            if ( workerCount > 1 && deploymentQueue.isEmpty() ) {
                delete = true;
            }
            else if ( workerCount == MAX_WORKERS ) {
                create = false;
            }
            else if ( workerCount == 0 ) {
                create = true;
            }
            else if ( !deploymentQueue.isEmpty() ) {
                create = true;
            }

            if ( create ) {
                Worker worker = new Worker();
                workers.add(worker);
                worker.start();
                log.info(String.format("Added deployment worker %d [Total=%d]", worker.getId(), workers.size()));
            }

            if ( delete ) {
                Iterator<Worker> iterator = workers.iterator();
                Worker worker = iterator.next();
                if ( worker != null ) {
                    worker.shutdown();
                    iterator.remove();
                }
                log.info(String.format("Deleted deployment worker %d [Total=%d]",worker.getId(), workers.size()));
            }
        }

        private void printQueueStats() {
            if ( deploymentQueue.isEmpty() ) {
                if ( start != 0 ) {
                    log.info(String.format("Deployment queue emptied after %d millis", System.currentTimeMillis()-start));
                    start = 0;
                }
            }
            else if ( start == 0 ) {
                start = System.currentTimeMillis();
            }
            else {
                log.info(String.format("Deployment queue length=%d after %d millis", deploymentQueue.size(), System.currentTimeMillis()-start));
            }
        }
    }

    private class Worker extends Thread {
        private final MashupDeployer deployer;

        public Worker() {
            deployer = newDeployer();
            log.debug(String.format("Worker %d constructed", this.getId()));
        }

        public void shutdown() {
            this.interrupt();
            log.debug(String.format("Worker %d shutting down", this.getId()));
        }

        @Override
        public void run() {
            log.debug( String.format("Worker %d entering run()",this.getId()) );

            try {
                while ( !isInterrupted() ) {
                    DeploymentFileData deploymentFileData = deploymentQueue.poll(Integer.MAX_VALUE, TimeUnit.DAYS);

                    try {
                        log.debug( String.format("Worker %d executing deploy()",this.getId()) );
                        deployer.deploy(deploymentFileData);
                    }
                    catch (DeploymentException e) {
                        log.warn(String.format("Exception in Worker %d : %s", this.getId(), e.getMessage()), e);
                    }
                }
            }
            catch (InterruptedException e) {}

            log.debug( String.format("Worker %d exiting run()", this.getId()) );
        }
    }


}
